package com.mw.job.zookeeper;

import com.mw.job.bean.AppProperties;
import com.mw.job.bean.DisruptorProperties;
import com.mw.job.bean.ZkProperties;
import com.mw.job.context.SpringContextUtil;
import com.mw.job.util.HelpMe;
import com.xiaoleilu.hutool.util.NetUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.web.ServerProperties;

import java.util.Map;

/**
 * Created by mengdexuan on 2017/10/12 16:25.
 */
public class SessionConnectionStateListener implements ConnectionStateListener{

	private static Logger log = LoggerFactory.getLogger(SessionConnectionStateListener.class);

	ServerProperties serverProperties = SpringContextUtil.getBean(ServerProperties.class);
	DisruptorProperties disruptorProperties = SpringContextUtil.getBean(DisruptorProperties.class);
	AppProperties appProperties = SpringContextUtil.getBean(AppProperties.class);
	ZkProperties zkProperties = SpringContextUtil.getBean(ZkProperties.class);

	@Override
	public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {

		//连接到 ZK 服务器后，将节点信息注册到 ZK 上
		if(ConnectionState.RECONNECTED==connectionState||ConnectionState.CONNECTED==connectionState){

			if (ConnectionState.RECONNECTED==connectionState){
				ZkClient.close();
				ZkClient.connect(zkProperties.getConnectString(),zkProperties.getSessionTimeout(),appProperties);
			}

			String ip = NetUtil.getLocalhostStr();

			try {
				Map<String, Object> map = HelpMe.convertBean(disruptorProperties);
				String mapStr = HelpMe.easyMap2Str(map);

				String tempContextPath = HelpMe.removeFirstChar(serverProperties.getContextPath());
				String zkNode = ip+":"+serverProperties.getPort()+":"+tempContextPath;

				String nodePath = "/MW-JOB/"+appProperties.getKey()+"/node";

				nodePath += "/";
				nodePath += zkNode;

				//将节点调度信息注册到 ZK 上
				ZkClient.createNode(nodePath,mapStr, CreateMode.EPHEMERAL);

				log.info("注册节点：nodePath --> {},nodeVal --> {}",new Object[]{nodePath,mapStr});

			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
